package com.atguigu.tingshu.search.stream;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.kafka.KafkaMessagePojo;
import com.atguigu.tingshu.kafka.KafkaStreamResultPojo;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.dao.AlbumInfoDao;
import java.time.Duration;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/****
 * kafka流式编程的任务: 统计用户的行为数据
 */
@Component
public class KafkaStreamTask {

    /**
     * 统计流式编程的业务逻辑
     * @param streamsBuilder
     * @return
     */
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder){
        //从哪个上游topic获取消息
        KStream<String, String> stream =
                streamsBuilder.stream("tingshu_stream_rank_topic");
        //对数据进行统计处理
        stream
                //将所有的消息先进行合并才能获取到一个集合
                .flatMapValues(message ->{
                    //反序列化
                    KafkaMessagePojo pojo =
                            JSONObject.parseObject(message, KafkaMessagePojo.class);
                    //返回: 1:play
                    return Arrays.asList(pojo.getAlbumId() + ":" + pojo.getType());
                }).map((key, value)->{
                    //将消息的key和value都换成: 1:play
                    return new KeyValue<>(value, value);
                })
                //才能根据集合中的数据,使用key进行groupBy分组
                .groupByKey()
                //定义时间窗口: 多久统计一次
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(10)))
                //具体的业务: count统计
                .count()
                .toStream()
                //将统计的数据拿出来进行转换发送下游
                .map((key, value)->{
                    //获取桶名字: 1:play
                    String keyString = key.key().toString(); // 1
                    return new KeyValue<>(keyString, getKafkaStreamResultPojo(keyString, value));
                }).to("tingshu_last_topic");
        //返回
        return stream;
    }

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private AlbumInfoDao albumInfoDao;
    /**
     * 下游消费者
     */
    @KafkaListener(topics = "tingshu_last_topic")
    public void tingshuLastTopicConsumer(ConsumerRecord<String, String> record){
        //获取到统计后的行为数据和数量
        KafkaStreamResultPojo pojo =
                JSONObject.parseObject(record.value(), KafkaStreamResultPojo.class);
        // 打印出数据
        // System.out.println("下游主题收到的内容为" + pojo);

    //获取专辑id
    Long albumId = pojo.getAlbumId();
    //获取一级分类id
    AlbumInfoIndex albumInfoIndex = albumInfoDao.findById(albumId).get();
    //获取一级分类的id
    Long category1Id = albumInfoIndex.getCategory1Id();
    //获取日期
    String time = new DateTime().toString("yyyyMMdd");
    //实时日排行榜
    Long playNum = pojo.getPlayNum();
    redisTemplate.opsForZSet().incrementScore(
            category1Id + ":playStatNum:" + time,
            albumId,
            -playNum);
    Long collectNum = pojo.getCollectNum();
    redisTemplate.opsForZSet().incrementScore(
            category1Id + ":subscribeStatNum:" + time,
            albumId,
            -collectNum);//收藏
    Long buyNum = pojo.getBuyNum();
    redisTemplate.opsForZSet().incrementScore(
            category1Id + ":buyStatNum:" + time,
            albumId,
            -buyNum);
    Long commentNum = pojo.getCommentNum();
    redisTemplate.opsForZSet().incrementScore(
            category1Id + ":commentStatNum:" + time,
            albumId,
            -commentNum);
    // 计算热度值
    Long hotScore = playNum + collectNum*2 + buyNum*10 + commentNum*5;
    redisTemplate.opsForZSet().incrementScore(
            category1Id + ":hotScore:" + time,
            albumId,
            -hotScore);
  }

    /**
     * 前端参数:  行为+ 一级分类id
     */

    /**
     * 后续处理逻辑
     * @param key: 1:play
     * @param value: 数量
     * @return
     */
    private String getKafkaStreamResultPojo(String key, Long value) {
        //初始化
        KafkaStreamResultPojo pojo = new KafkaStreamResultPojo();
        //key进行切分
        String[] split = key.split(":");
        //获取主体的id
        Long albumId = Long.valueOf(split[0]);
        pojo.setAlbumId(albumId);
        switch (split[1]){
            //保存播放量
            case "play" -> pojo.setPlayNum(value);
            case "comment" -> pojo.setCommentNum(value);
            case "buy" -> pojo.setBuyNum(value);
            case "collect" -> pojo.setCollectNum(value);
        }
        return JSONObject.toJSONString(pojo);
    }}
